package cn.zmwh.im.server.mq;

import cn.hutool.core.collection.CollUtil;
import cn.zmwh.im.common.pojo.dto.IMRecvInfo;
import cn.zmwh.im.server.mq.handler.MessageSendHandler;
import cn.zmwh.im.server.netty.IMServerGroup;
import cn.zmwh.im.server.utils.ThreadPoolExecutorFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * reids 队列拉取定时任务
 *
 * @author: dmzmwh
 * @date: 2024-07-15
 * @version: 1.0
 */
@Slf4j
@Component
public class MessageMQPullTask implements CommandLineRunner {

    private static final ScheduledThreadPoolExecutor EXECUTOR = ThreadPoolExecutorFactory.getThreadPoolExecutor();

    @Autowired(required = false)
    private List<MessageSendHandler> consumers = Collections.emptyList();

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private IMServerGroup serverGroup;

    @Value("${pull_one:false}")
    private boolean pullOne;

    @Override
    public void run(String... args) {
        consumers.forEach((consumer -> {
            String queue = consumer.getQueue();
            int batchSize = consumer.getBatchSize();
            int period = consumer.getInterval();
            EXECUTOR.execute(new Runnable() {
                @Override
                public void run() {
                    int reqSize = 0;
                    try {
                        if (serverGroup.isReady()) {
                            String key = queue + ":" + IMServerGroup.serverId;
                            // 拉取一个批次的数据
                            List<IMRecvInfo> list = pullBatch(key, batchSize);
                            if (CollUtil.isNotEmpty(list)) {
                                reqSize = list.size();
                                for (IMRecvInfo obj : list) {
                                    log.info("数据消费,队列:{},数据:{}", queue, obj.toString());
                                    consumer.onMessage(obj);
                                }
                            }
                        }
                    } catch (Exception e) {
                        log.error("数据消费异常,队列:{}", queue, e);
                        // 出现异常，10s后再重新尝试消费
                        EXECUTOR.schedule(this, 10, TimeUnit.SECONDS);
                        return;
                    }
                    // 继续消费数据
                    if (!EXECUTOR.isShutdown()) {
                        if (reqSize < batchSize) {
                            // 数据已经消费完，等待下一个周期继续拉取
                            EXECUTOR.schedule(this, period, TimeUnit.MILLISECONDS);
                        } else {
                            // 数据没有消费完，直接开启下一个消费周期
                            EXECUTOR.execute(this);
                        }
                    }
                }
            });
        }));
    }

    private List<IMRecvInfo> pullBatch(String key, Integer batchSize) {
        List<Object> objects = new ArrayList<>();
        if (pullOne) {
            int index = 0;
            do {
                index++;
                // 低版本Redis，只能逐条拉取
                Object obj = redisTemplate.opsForList().leftPop(key);
                if (obj != null) {
                    objects.add(obj);
                }
            } while (objects.size() < batchSize && index < batchSize);

        } else {
            objects = redisTemplate.opsForList().leftPop(key, batchSize);
        }
        if (CollUtil.isEmpty(objects)) {//没有拉取到数据
            return Collections.emptyList();
        }
        List<IMRecvInfo> collect = objects.stream().map(obj -> (IMRecvInfo) obj).collect(Collectors.toList());

        return collect;
    }

    @PreDestroy
    public void destory() {
        log.info("消费线程停止...");
        ThreadPoolExecutorFactory.shutDown();
    }
}
